package com.gitee.johnvwan.rocketmq;

import com.gitee.johnvwan.rocketmq.core.AbstractDeDupConcurrentListener;
import com.gitee.johnvwan.rocketmq.core.DedupConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.data.redis.core.StringRedisTemplate;

@Slf4j
public class SampleListener extends AbstractDeDupConcurrentListener {

    public SampleListener(DedupConfig dedupConfig) {
        super(dedupConfig);
    }

    public static void main(String[] args) throws MQClientException {


        //利用Redis做幂等表
        {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP1");
            consumer.subscribe("TEST-TOPIC", "*");

            String appName = consumer.getConsumerGroup();
            StringRedisTemplate stringRedisTemplate = null;
            // 这里省略获取StringRedisTemplate的过程
            DedupConfig dedupConfig = DedupConfig.enableDeDupConsumeConfig(appName, stringRedisTemplate);
            AbstractDeDupConcurrentListener messageListener = new SampleListener(dedupConfig);

            consumer.registerMessageListener(messageListener);
            consumer.start();
        }

        //利用MySQL做幂等表
        // {
        //     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP2");
        //     consumer.subscribe("TEST-TOPIC", "*");
        //     String appName = consumer.getConsumerGroup();
        //     // 这里省略获取JDBCTemplate的过程
        //     JdbcTemplate jdbcTemplate = null;
        //     DedupConfig dedupConfig = DedupConfig.enableDeDupConsumeConfig(appName, jdbcTemplate);
        //     DedupConcurrentListener messageListener = new SampleListener(dedupConfig);
        //
        //     consumer.registerMessageListener(messageListener);
        //     consumer.start();
        // }
    }

    /**
     * 基于什么做消息去重，每一类不同的消息都可以不一样，做去重之前会尊重此方法返回的值
     *
     * @param messageExt 消息ext
     * @return {@link String}
     */
    @Override
    protected String deDupMessageKey(MessageExt messageExt) {
        //为了简单示意，这里直接使用消息体作为去重键
        if ("TEST-TOPIC".equals(messageExt.getTopic())) {
            return new String(messageExt.getBody());
        } else {
            //其他使用默认的配置（消息id）
            return super.deDupMessageKey(messageExt);
        }
    }

    /**
     * 做处理消息
     *
     * @param messageExt 消息ext
     * @return boolean
     */
    @Override
    protected boolean doHandleMsg(MessageExt messageExt) {
        switch (messageExt.getTopic()) {
            case "TEST-TOPIC":
                log.info("假装消费很久....{} {}", new String(messageExt.getBody()), messageExt);
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                }
                break;
        }
        return true;
    }

}
